package rocketmq

import (
	"context"
	"strconv"
	"strings"
	"sync"
	"time"

	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/internal"

	"github.com/gogap/errors"
)

const (
	transUnknown = iota
	transCommit
	transRollback
)

func (s *Subscriber) subscribeAliyunTrans(ctx context.Context) {
	var wg sync.WaitGroup

	respChan := make(chan internal.ConsumeMessageResponse)
	errChan := make(chan error)

	// 订阅事务二阶段确认
	p, err := s.client.CreateTransProducer(s.topicName, s.groupID)
	if err != nil {
		s.logger.Errorc(ctx, "subscribeAliyunTrans: CreateTransProducer[%s] [%s]创建失败[%s]", s.topicName, s.groupID)
		return
	}
	// 注册监听
	s.client.RegisterListener(s.topicName, s)

	wg.Add(1)
	go func(ctx context.Context) {
		defer wg.Done()
		select {
		case <-ctx.Done():
			s.logger.Infof("ConsumeHalfMessage 读取消息超时，返回重试；")
		case <-time.After(40 * time.Second):
			s.logger.Infof("ConsumeHalfMessage 读取消息超时40s;")
		case resp := <-respChan: // 生产者两阶段事务消息确认
			for i := 0; i < len(resp.Messages); i++ {
				v := internal.ConsumeMessage{Messages: resp.Messages[i]}

				s.logger.Infof("Receive MessageID: [%s], PublishTime: [%d], MessageTag: [%s] ConsumedTimes: [%d], FirstConsumeTime: [%d], NextConsumeTime: [%d]",
					v.Messages.MessageId, v.Messages.PublishTime, v.Messages.MessageTag, v.Messages.ConsumedTimes, v.Messages.FirstConsumeTime, v.Messages.NextConsumeTime)

				// rocketMQ 两阶段事务确认
				ack, err := s.handler(&v)
				if err != nil {
					s.logger.Errorf("两阶段事务确认 处理失败[%s][%s]下次接收时间[%d]",
						v.Messages.MessageId, v.Messages.MessageBody, v.Messages.NextConsumeTime)
					continue
				}

				switch ack.Transaction {
				case transUnknown: // 不处理
					s.logger.Infof("Transaction get UnKnown result, continue!")
					continue
				case transCommit: // 提交
					err = p.Commit(v.Messages.ReceiptHandle)
				case transRollback: // 回滚
					err = p.Rollback(v.Messages.ReceiptHandle)
				default:
					s.logger.Errorf("caught unknown flag[%d], continue!", ack.Transaction)
					continue
				}
				s.logger.Infof("ConsumeHalfMessage: Ack[%s] ---->[%s]确认结果[%v];", v.Messages.MessageId, ack.ReceiptHandle, err)
				s.debugError(err)
			}

		case err := <-errChan: // 订阅消息失败
			if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
				s.logger.Debugf("ConsumeHalfMessage: No new message, continue")
				return
			}
			s.logger.Errorf("ConsumeHalfMessage 读取事务确认处理[%s]", err)
			s.debugError(err)
		}
	}(ctx)

	go func() {
		p.ConsumeHalfMessage(respChan, errChan, int32(s.messageNum), int64(s.maxWait))
	}()

	wg.Wait()
}

func (s *Subscriber) subscribeApacheTrans() {
	// 注册Listener
	s.client.RegisterListener(s.topicName, s)
	// 创建事务Producer
	_, err := s.client.CreateTransProducer(s.topicName, s.groupID)
	if err != nil {
		s.logger.Panicf("subscribeApacheTrans: CreateTransProducer[%s] [%s]创建失败[%s]", s.topicName, s.groupID)
	}
}

// ExecuteLocalTransaction When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
func (s *Subscriber) ExecuteLocalTransaction(msg *internal.Message) internal.LocalTransactionState {
	s.logger.Infof("两阶段消息确认处理:[%s]", msg.String())
	v := internal.ConsumeMessage{
		Messages: internal.ConsumeMessageEntry{
			MessageBody: string(msg.Body),
			Properties:  msg.GetProperties(),
		},
	}

	v.Messages.StartDeliverTime, _ = strconv.ParseInt(msg.GetProperty(internal.StartDeliverTime), 10, 64)
	v.Messages.TransCheckImmunityTime, _ = strconv.Atoi(msg.GetProperty(internal.TransCheckImmunityTime))
	v.Messages.ShardingKey = msg.GetProperty(internal.SHARDING)

	// rocketMQ 两阶段事务确认
	ack, err := s.handler(&v)
	if err != nil {
		s.logger.Errorf("两阶段事务确认 处理失败[%s][%s]下次接收时间[%d]",
			v.Messages.MessageId, v.Messages.MessageBody, v.Messages.NextConsumeTime)
		return internal.UnknownState
	}

	switch ack.Transaction {
	case transUnknown: // 不处理
		s.logger.Infof("Transaction get UnKnown result, continue!")
		return internal.UnknownState
	case transCommit: // 提交
		return internal.CommitMessageState
	case transRollback: // 回滚
		return internal.RollbackMessageState
	}
	return internal.UnknownState
}

// CheckLocalTransaction When no response to prepare(half) message. broker will send check message to check the transaction status, and this
// method will be invoked to get local transaction status.
func (s *Subscriber) CheckLocalTransaction(msg *internal.MessageExt) internal.LocalTransactionState {
	/*此处反查状态不确认*/
	return internal.UnknownState
}
